package cn.itcast.streaming.flatmap;

import cn.itcast.streaming.entity.ElectricFenceModel;
import cn.itcast.streaming.entity.ElectricFenceResultTmp;
import cn.itcast.streaming.entity.ItcastDataPartObj;
import cn.itcast.streaming.utils.DateUtil;
import cn.itcast.streaming.utils.DistanceCaculateUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.streaming.api.functions.co.RichCoFlatMapFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;

/**
 * Author itcast
 * Date 2021/9/26 11:50
 * Desc 将拉宽的实时车辆数据和电子围栏规则数据flatMap生成新对象 ElectricFenceModel
 */
public class ElectricFenceRulesFuntion extends RichCoFlatMapFunction<ItcastDataPartObj, HashMap<String, ElectricFenceResultTmp>, ElectricFenceModel> {

    private static final Logger logger = LoggerFactory.getLogger(ElectricFenceRulesFuntion.class);

    //定义接收电子栅栏配置数据的变量
    private static HashMap<String, ElectricFenceResultTmp> electricFenceConfig = null;

    @Override
    public void flatMap1(ItcastDataPartObj itcastDataPartObj, Collector<ElectricFenceModel> out) throws Exception {
        //1.定义返回的 ElectricFenceModel
        ElectricFenceModel electricFenceModel = new ElectricFenceModel();
        //2.判断如果流数据数据质量（车辆的经纬度不能为0或-999999，车辆GpsTime不能为空）
        if (isValid(electricFenceModel.getLng()) && isValid(electricFenceModel.getLat())) {
            //2.1.获取当前车辆的 vin
            String vin = electricFenceModel.getVin();
            //2.2.通过vin获取电子围栏的配置信息
            ElectricFenceResultTmp electricFence = electricFenceConfig.getOrDefault(vin, null);
            //2.3.如果电子围栏配置信息不为空
            if (electricFence != null) {
                //2.3.1.说明当前车辆关联了电子围栏规则，需要判断当前上报的数据是否在电子围栏规则的生效时间内，
                // 先获取上报地理位置时间gpsTimestamp
                long startTimeStamp = electricFence.getStartTime().getTime();
                long endTimeStamp = electricFence.getEndTime().getTime();
                //当前车辆上报的实时 gps 时间戳
                long gpsTimeStamp = DateUtil.convertStringToDateTime(itcastDataPartObj.getGpsTime()).getTime();
                //2.3.2.如果当前gpsTimestamp>=开始时间戳并且gpsTimestamp<=结束时间戳，以下内容存入到 ElectricFenceModel
                if (gpsTimeStamp >= startTimeStamp && gpsTimeStamp <= endTimeStamp) {
                    //2.3.2.1.上报车辆的数据在电子围栏生效期内 vin gpstime lng lat 终端时间和终端时间戳
                    electricFenceModel.setVin(itcastDataPartObj.getVin());
                    electricFenceModel.setGpsTime(itcastDataPartObj.getGpsTime());
                    electricFenceModel.setLng(itcastDataPartObj.getLng());
                    electricFenceModel.setLat(itcastDataPartObj.getLat());
                    electricFenceModel.setTerminalTime(itcastDataPartObj.getTerminalTime());
                    electricFenceModel.setTerminalTimestamp(itcastDataPartObj.getTerminalTimeStamp());
                    //2.3.2.2.电子围栏id，电子围栏名称，地址，半径
                    electricFenceModel.setEleId(electricFence.getId());
                    electricFenceModel.setEleName(electricFence.getName());
                    electricFenceModel.setAddress(electricFence.getAddress());
                    electricFenceModel.setRadius(electricFence.getRadius());
                    //2.3.2.3.电子围栏经纬度
                    electricFenceModel.setLongitude(electricFence.getLongitude());
                    electricFenceModel.setLatitude(electricFence.getLatitude());
                    //2.3.2.4.计算经纬度和电子围栏经纬度距离距离，如果两点之间大于半径（单位是千米）的距离，
                    // 就是存在于圆外，否则反之
                    Double distance = DistanceCaculateUtil.getDistance(itcastDataPartObj.getLat(), itcastDataPartObj.getLng()
                            , electricFence.getLatitude(), electricFence.getLongitude());
                    if (distance / 1000 > electricFence.getRadius()) {
                        electricFenceModel.setNowStatus(1);
                    } else {
                        electricFenceModel.setNowStatus(0);
                    }
                    //2.3.2.5.收集结果数据
                    out.collect(electricFenceModel);
                } else {
                    logger.warn("当前电子栅栏的时间无效");
                }
            } else {
                logger.warn("当前车辆%s对应电子栅栏不存在", vin);
            }
        } else {
            logger.warn("当前车辆数据上报的经度%s和维度%s不能为空", electricFenceModel.getLng(), electricFenceModel.getLat());
        }
    }

    /**
     * 如果满足不等于0并且-999999，说明数据有效，如果有效为 true 否则为 false
     *
     * @param value
     * @return
     */
    private boolean isValid(Double value) {
        if (StringUtils.isNotBlank(value + "")) {
            if (value != 0 && value != -999999) {
                return true;
            } else {
                return false;
            }
        } else {
            return false;
        }
    }

    @Override
    public void flatMap2(HashMap<String, ElectricFenceResultTmp> value, Collector<ElectricFenceModel> out) throws Exception {
        electricFenceConfig = value;
    }
}
